package container

import (
	"context"
	"fmt"
	"time"

	"gitee.com/doraemon1996/bunny/goroutine"
	"gitee.com/doraemon1996/bunny/log"
	"gitee.com/doraemon1996/bunny/structure"
)

// 容器结构
type container struct {
	structure.Control

	// 容器ID
	containerID uint64

	// 服务对象
	instance Service

	// 上次活跃时间 没有消息处理时, 每秒更新一次
	activeTime time.Time

	// 消息处理队列
	inPipe chan Message
}

func (ctn *container) Init(ctx context.Context, id uint64, srv Service, args ...any) (err error) {
	ctn.Control.Init(ctx, func() {
		err = srv.Init(ctn.Context(), id, args...)
		if err != nil {
			return
		}

		ctn.containerID = id
		ctn.instance = srv
		ctn.inPipe = make(chan Message, 10000)
	})
	return
}

func (ctn *container) Start() (err error) {
	ctn.Control.Start(func() {
		err = ctn.instance.Start()
		if err != nil {
			return
		}

		ctn.activeTime = time.Now()
		ctn.Add(1)
		goroutine.Submit(func() {
			defer ctn.Done()
			ctn.dispatch()
		})
	})
	return
}

func (ctn *container) Stop() {
	ctn.Control.Stop(func() {
		ctn.Wait()

		close(ctn.inPipe)
		ctn.instance.Stop()

		ctn.inPipe = nil
		ctn.instance = nil
	})
}

func (ctn *container) ContainerID() uint64 {
	return ctn.containerID
}

func (ctn *container) String() string {
	return fmt.Sprintf("Container[%X]", ctn.containerID)
}

func (ctn *container) ActiveCheck(now time.Time) bool {
	return now.Sub(ctn.activeTime) < time.Second*5
}

func (ctn *container) PutMessage(msg Message) {
	if ctn.Opening() {
		ctn.inPipe <- msg
	} else {
		log.Errorf("PutMessage but contaier closed. msg: %s", msg)
		msg.Error(ErrContainerClosed)
	}
}

// 安全调用消息处理函数
func (ctn *container) invoke(msg Message) {
	defer func() {
		if val := recover(); val != nil {
			err := fmt.Errorf("%v", val)
			log.Errorf("Container[%X] handler message failure. err: %v", ctn.containerID, err)
			msg.Error(err)
		}
	}()

	start := time.Now()
	ctn.instance.OnMessage(msg)
	if sub := time.Since(start); sub > time.Second {
		log.Warnf("Container[%X] cost %s to dispatch message[%d] from %X", ctn.containerID, sub.String(), msg.Data().MsgId, msg.Data().Source)
	}
}

// 消息处理循环
func (ctn *container) dispatch() {
	ticker := time.NewTicker(time.Second)
	defer ticker.Stop()

	for {
		ctn.activeTime = time.Now()
		select {
		case <-ctn.CtxDone():
			return
		case msg := <-ctn.inPipe:
			ctn.invoke(msg)
		case <-ticker.C:
		}
	}
}
